package com.shujia.spark.streaming

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Durations, StreamingContext}

import java.util.Properties

object Demo09OnKafkaByStreaming {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.replace("$", ""))
      .master("local[2]") // 设置运行的方式
      .getOrCreate()

    val ssc = new StreamingContext(spark.sparkContext, Durations.seconds(5))

    val kafkaParams: Map[String, Object] = Map[String, Object](
      "bootstrap.servers" -> "master:9092,node1:9092,node2:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "group01",
      "auto.offset.reset" -> "earliest",
      "enable.auto.commit" -> (false: java.lang.Boolean)
    )

    val topics: Array[String] = Array("spark_t1")
    val stream: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      ssc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )


    val wordCntTopic: String = "word_cnt_01"

    stream
      .flatMap(record => record.value.split(","))
      .map(w => (w, 1))
      .reduceByKey(_ + _)
      // 保存到Kafka
      .foreachRDD(rdd => {
        rdd.foreachPartition(it => {
          val prop: Properties = new Properties
          // 设置Kafka Broker的地址// 设置Kafka Broker的地址
          prop.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")
          prop.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
          prop.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
          val kafkaProducer: KafkaProducer[String, String] = new KafkaProducer[String, String](prop)

          it.foreach(kv => {
            kafkaProducer.send(new ProducerRecord(wordCntTopic, s"${kv._1},${kv._2}"))
          })

          kafkaProducer.flush()
        })
      })

    ssc.start()
    ssc.awaitTermination()
  }

}
